package mrt.cse.tempworks.initialWorksOfTeamMembers.sumedha;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class ClientHandler implements org.apache.zookeeper.Watcher {

    protected ZooKeeper zk;
    private static final int SESSION_TIMEOUT = 5000;
    protected boolean expired;
    protected String host;
    protected String zNode;
    protected int numberOfManagerNodes = 0;
    protected HashMap<Long,String> joinedManagers;

    protected String managerName;
    protected long zxid;
    protected int version;
    public boolean isLeader;

    private CountDownLatch connectedSignal = new CountDownLatch(1);

    /**
     * Connects to the ZooKeeper Service
     * @param hosts address of the ZooKeeper Server
     * @throws IOException
     * @throws InterruptedException
     * @throws KeeperException
     */
    public void connect(String hosts) throws IOException, InterruptedException, KeeperException {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
        connectedSignal.await();
    }

    /**
     * Assigns values for the managers fields
     * zxid,versions
     * This is done when the Manager oblect is created
     */
    public void assignValues() {
        try{
            Stat stat = zk.exists("/" + zNode + "/" + managerName, false);
            zxid = stat.getCzxid();
            version = stat.getVersion();
            updateManagerList();
            numberOfManagerNodes = zk.getChildren("/" + zNode ,true).size();
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    /**
     * Each manager keeps a list of all the connected managers to the ZooKeeper server. When a
     * new manager joins or a manager leaves, thi method will update the above list of all
     * managers.
     */
    public void updateManagerList(){
        joinedManagers = new HashMap<Long, String>();
        try {
            List<String> tempList = zk.getChildren("/" + zNode, false);
            System.out.println(tempList);
            for(String managerName : tempList){
                Stat tempStat = zk.exists("/" + zNode + "/" + managerName, false);
                joinedManagers.put(tempStat.getCzxid(),managerName);
            }
            if(joinedManagers.size() == 1){
                isLeader = true;
                System.out.println("Leader is : " + managerName);
                writeLeaderIntoParentNode();
            }
            else{
                electLeader();
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(joinedManagers);
    }

    /**
     * Joins with the ZooKeeper
     * @param groupName  Name of The Node which is intended to join
     * @param managerName  Name for the node created under group
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void join(String groupName, String managerName) throws KeeperException,
            InterruptedException {
        String path = "/" + groupName + "/" + managerName;
        String createdPath;
        createdPath = zk.create(path, this.managerName.getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL);
        System.out.println("Created " + createdPath);
    }


    public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.None) {
            switch (event.getState()) {
                case SyncConnected:
                    System.out.println("Connected to the ZooKeeper Service");
                    connectedSignal.countDown();
                    break;
                case Expired:
                    expired = true;
            }
        }
        //Watches for changes in the child nodes i.e. Create or Delete
        else if(event.getType() == Event.EventType.NodeChildrenChanged) {
            try {
                int currentNoOfNodes = zk.getChildren("/" + zNode ,false).size();
                if(currentNoOfNodes <= numberOfManagerNodes){
                    System.out.println("Manager Has Left");
                    updateManagerList();
                }
                else if(currentNoOfNodes > numberOfManagerNodes){
                    System.out.println("Manager Has Joined");
                    updateManagerList();
                }
                numberOfManagerNodes = currentNoOfNodes;
                watchNode();
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void close() throws InterruptedException {
        zk.close();
    }

    /**
     * Puts a Watcher on /zoo for node creation and deletion
     */
    private void watchNode() {
        try {
            zk.getChildren("/" + zNode ,true);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * Elects the leader according to the Zxid
     * i.e. The manager with the minimum zxid will be elected as the leader
     * The leader name will be saved in data of the Parent (Group) node
     */
    private void electLeader() {

        Long[] zxidArray = new Long[joinedManagers.size()];
        int i = 0;
        for(Long zxid : joinedManagers.keySet()){
            zxidArray[i] = zxid;
            i++;
        }
        Arrays.sort(zxidArray);
        long leaderZxid = zxidArray[0];
        System.out.println("Leader is : " + joinedManagers.get(leaderZxid));
        if(leaderZxid == zxid){
            isLeader = true;
            writeLeaderIntoParentNode();
        }
        else{
            isLeader = false;
        }
    }

    /**
     * Writes the name of the Leader into the Parent Node
     */
    private void writeLeaderIntoParentNode() {
        byte[] bytes = managerName.getBytes();
        try {
            zk.setData("/" + zNode , bytes ,-1);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


}